package com.niit.rdd

import org.apache.spark.{SparkConf, SparkContext}

object Spark_RDD_Transform03 {
  def main(args: Array[String]): Unit = {
    //准备环境
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("spark")
    val sc = new SparkContext(sparkConf)
    sc.setLogLevel("ERROR")
    /*
    groupByKey:将数据源根据Key对Value进行分组
     */

    val rdd1 = sc.makeRDD( List( ("a",1) , ("a",2),("a",3),("b",4) ) )
    val groupRDD =  rdd1.groupByKey();
    //结果是元组 元组中的第一个元素是key，元素第二个元素就是相同可以的value集合
    groupRDD.collect().foreach( println )

    println("----------------------------------------")

    /*
      aggregateByKey:将数据根据不同的规则进行分区内和分区间计算
                ByKey:根据Key来去操作
       1 【 (a,1) (a,2) (b,3) 】==>  (a,2) (b,3)  分区内最大值
       2 【  (b,4) (b,5) (a,6) 】 ==>   (b,5) (a,6)  分区内最大值
       分区间的最大值和
         a  2+6
         b  3+5
         (a,8)
         (b,8)
     */
    val rdd2 =  sc.makeRDD( List( ("a",1),("a",2),("b",3),
                                  ("b",4),("b",5),("a",6)
    ),2)

    /*
    参数1：表示初始值
    参数2：需要传递两个参数  第一个分区内计算规则，第二个分区间计算规则
     */
    rdd2.aggregateByKey(0)(
      (x,y)=>math.max(x,y),
      (x,y)=>x+y
    ).collect().foreach( println )
    /*
    第一步：分区内找最大值
        【 (a,1) (a,2) (b,3),(a,0),(b,0) 】==>   a 0,1,2  (a,2) b 0,3 (b,3)
        【  (b,4) (b,5) (a,6),(a,0),(b,0) 】 ==>   a 0,6  (a,6) b 0,4,5  (b,5)
    第二步 分区间计算
        【(a,2) (a,6)】   【(b,3)， (b,5)】
        key：a                 b
        value:x=2 y=6 x+y      x=3,y=5  x+y
        (a,8)           (b,8)
     */
    println("--------------------------------------")
    //需求：求两个分区间的总和
    val rdd3 =  sc.makeRDD( List( ("a",1),("a",2),("b",3),
      ("b",4),("b",5),("a",6)
    ),2)

    rdd3.aggregateByKey(5)(
      (x,y) => x+y ,
      (x,y)=> x+y
    ).collect().foreach(println)
  /*
    第一步：分区内的和
    【 (a,1) (a,2) (b,3),(a,5),(b,5) 】==>   a 5,1,2  (a,8)  b 5,3 (b,8)
    【  (b,4) (b,5) (a,6),(a,5),(b,5) 】 ==>   a 5,6  (a,11) b 5,4,5  (b,14)
    第二步 分区间计算
    【(a,8) (a,11)】   【(b,8)， (b,14)】
    key：a                     b
      value:x=8 y=11 x+y      x=8,y=14  x+y
      (a,19)           (b,22)
    */
  //简写方式：
    rdd3.aggregateByKey(0)(_+_ , _+ _ ).collect().foreach(println)
  //再简写方式:当分区内和分区间的计算规则相同时可以使用foldByKey
    rdd3.foldByKey(0)(_ + _).collect().foreach(println)

    /*
    sortBykye:根据key来进行排序
     */
    val rdd4 = sc.makeRDD( List(("a",1) , ("c",3) , ("b",2)) )
    //==>（a,1） (b,2) (c,3)
    val sortKey =  rdd4.sortByKey(false);
    sortKey.collect().foreach(println)

    /*
      join  leftOuterJoin RightOuterJoin
     */
    println("----------------------")
    val rdd5 = sc.makeRDD( List(  ("a",1) ,("a",2) ,("c",3),("b",8) ) )
    val rdd6 = sc.makeRDD( List(  ("a",5) ,("c",6) ,("a",4),("g",7) ) )
    val joinRdd =  rdd5.join(rdd6)
    joinRdd.collect().foreach(println)
    println("----------------------")
    val joinRdd2 = rdd5.leftOuterJoin(rdd6)
    joinRdd2.collect().foreach(println)
    val joinRdd3 = rdd5.rightOuterJoin(rdd6)
    joinRdd3.collect().foreach(println)

    /*
      aggregateByKey 计算平均值
     */

    val rdd7 =  sc.makeRDD( List( ("a",1),("a",2),("b",3),
                                   ("b",4),("b",5),("a",6)
    ),2)
            //(0,0)==(t,v):第一个0 表示的 v 的初始值 a:0+1+2  b:0+3
            //(0,0)==(t,v):第二个0 表示的 key出现的次数 a:0+1+1  b:0+1
            //t1 :第一个分区  a=== (3,2)    b=== (3,1 )
            //t2: 第二个分区   a=== (6,1)    b===(9,2)
            //t1表示第一个分区   t2表示第二个分区
    val newRdd = rdd7.aggregateByKey( (0,0) )(
      (t,v)=>{
        (t._1+v,t._2+1)
      },
      (t1,t2)=>{
        //a:v的相加：3+6 9        key出现次数相加 2 + 1  3
        //b:v的相加：3+9 12        key出现次数相加 1 + 2 3
        (t1._1+t2._1,             t1._2+t2._2)
      }
    )
    // newRdd  a:(9,3)  b:(12,3)

   val resultRdd =  newRdd.mapValues{
         // 9   , 3
         //12   , 3
      case (num,cnt)=>{
        num/cnt
        //9 / 3
        //12 / 3
      }
       //a = 3  b=4
    }
    resultRdd.collect().foreach(println)

    /*
        cogroup:分组后拼接  （connect + group）
     */
   val rdd8 =  sc.makeRDD( List( ("a",1),("b",2) ,//("c",3)
    )  )
    val rdd9 =  sc.makeRDD( List( ("a",4),("b",5) ,("c",6),("c",7)
    )  )
    /*
    a: 1,4
    b: 2,5
    c: 6,7
     */
    val cgRdd = rdd8.cogroup(rdd9)
    cgRdd.collect().foreach(println)


    sc.stop()





  }
}
